package cn.yanceysong.cdc;

import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadLocalRandom;

public class RedisKVMapper implements RedisMapper<String> {
    private static final Logger log = LoggerFactory.getLogger(RedisKVMapper.class);

    @Override
    public RedisCommandDescription getCommandDescription() {
        // 使用 Redis 的 SET 命令
        return new RedisCommandDescription(RedisCommand.SET);
    }

    @Override
    public String getKeyFromData(String data) {
        System.out.println(data);
        // 设置 Redis 的 Key
        return "mysql:bin-log:" + getRandomString(5);
    }

    @Override
    public String getValueFromData(String data) {
        log.info("receive data: {}", data);
        // 设置 Redis 的 Value
        return data;
    }
    public static String getRandomString(int length) {
        String base = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789";
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < length; i++) {
            int number = ThreadLocalRandom.current().nextInt(base.length());
            sb.append(base.charAt(number));
        }
        return sb.toString();
    }
}
